package cn.jia.activemq;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.junit.Test;

import javax.jms.*;

/**
 * @Author: jia
 * @Descirption:
 * @Date: 2018/4/19 14:02
 * @Modified By:
 */


public class TestActiveMQ {
    @Test
    //queue生产者
    public void testQueueProducer() throws JMSException {
        //1.创建连接工厂
        ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        //2创建一个连接对象
        Connection connection = factory.createConnection();
        //3.开启连接，调用start方法
        connection.start();
        //4.创建session,第一个参数：是否开启activeMQ事务，第二个参数：如果第一个为true，第二个参数自动忽略
        //如果我false，第二参数表示消息的应答模式。一般是自动应答。
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //5.使用session创建目的地：两种形式：queue/topic
        //参数为消息队列名称
        Queue queue = session.createQueue("queue");
        //6.使用session创建一个producer对象
        MessageProducer producer = session.createProducer(queue);
        //7.创建TextMessage对象
        TextMessage textMessage = new ActiveMQTextMessage();
        textMessage.setText("hello");
        //8.发送消息
        producer.send(textMessage);
        //9.关闭资源
        producer.close();
        session.close();
        connection.close();
    }

    @Test
    //queue发布者
    public void testQueueConsumer()throws Exception{
        //创建一个连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        //使用连接工厂创建一个连接
        Connection connection = connectionFactory.createConnection();
        //开启连接
        connection.start();
        //创建session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //创建接收的发源地
        Queue queue = session.createQueue("queue");
        //使用session创建Consumer
        MessageConsumer consumer = session.createConsumer(queue);
        //向Consumer中设置一个MessageListner,首先启动消费者，然后启动生产者，当生产者发布了消息后，
        //消费者会立马接收到
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                //取消息内容
                if(message instanceof TextMessage){
                    TextMessage textMessage= (TextMessage) message;
                    try {
                        String text = textMessage.getText();
                        System.out.println(text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        //系统等待接收消息
        /*while(true){
            Thread.sleep(100);
        }*/
        //等待接收输入
        System.in.read();
        consumer.close();
        session.close();
        connection.close();
    }
    //topic生产者,topic默认不会持久化到服务端，queue默认持久化到服务端
    @Test
    public void testTopicProducer() throws JMSException {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        //创建连接
        Connection connection = connectionFactory.createConnection();
        //开启连接
        connection.start();
        //创建session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //创建目的地
        Topic topic = session.createTopic("topic");
        //创建producer
        MessageProducer producer = session.createProducer(topic);
        //创建TextMessage
        TextMessage textMessage = session.createTextMessage("hello topic");
        producer.send(textMessage);
        //关闭
        producer.close();
        session.close();
        connection.close();
    }
    //topic消费者
    @Test
    public void testTopicConsumer1() throws Exception{
        //创建连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        //创建连接
        Connection connection = connectionFactory.createConnection();
        //开启连接
        connection.start();
        //创建session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //创建目的地
        Topic topic = session.createTopic("topic");
        //创建consumer
        MessageConsumer consumer = session.createConsumer(topic);
        //创建TextMessage
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                if(message instanceof TextMessage){
                    try {
                        String text = ((TextMessage) message).getText();
                        System.out.println("消费者1"+text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        System.in.read();
        //关闭
        consumer.close();
        session.close();
        connection.close();
    }
    //topic消费者
    @Test
    public void testTopicConsumer2() throws Exception{
        //创建连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
        //创建连接
        Connection connection = connectionFactory.createConnection();
        //开启连接
        connection.start();
        //创建session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //创建目的地
        Topic topic = session.createTopic("topic");
        //创建consumer
        MessageConsumer consumer = session.createConsumer(topic);
        //创建TextMessage
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                if(message instanceof TextMessage){
                    try {
                        String text = ((TextMessage) message).getText();
                        System.out.println("消费者2"+text);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        System.in.read();
        //关闭
        consumer.close();
        session.close();
        connection.close();
    }


    //subcriber：发送的消息可持久化到服务端

}
